package com.example.dayu521.msg_sender.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.dayu521.msg_sender.entity.Event;
import com.example.dayu521.msg_sender.entity.EventTasklist;
import com.example.dayu521.msg_sender.entity.SeckillSession;
import com.example.dayu521.msg_sender.mapper.EventMapper;
import com.example.dayu521.msg_sender.mapper.EventTasklistMapper;
import com.example.dayu521.msg_sender.mapper.LockMapper;
import com.example.dayu521.msg_sender.mapper.SeckillSessionMapper;
import com.example.dayu521.msg_sender.service.IEventTasklistService;
import com.example.dayu521.msg_sender.service.ISeckillSessionService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.example.dayu521.contants.AmqpC;
import org.example.dayu521.contants.RedisC;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;


/**
 * <p>
 * 秒杀活动场次 服务实现类
 * </p>
 *
 * @author baomidou
 * @since 2023-03-20
 */
@Service
@Slf4j
public class SeckillSessionServiceImpl extends ServiceImpl<SeckillSessionMapper, SeckillSession> implements ISeckillSessionService {


    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "redisTemplate")
    private RedisTemplate<String, Long> redisTemplate;

    @Resource(name = "redisTemplate")
    private RedisTemplate<String, String> lock;

    @Resource
    private LockMapper lockMapper;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private EventMapper eventMapper;

    @Resource
    private EventTasklistMapper eventTasklistMapper;

    @Resource
    private IEventTasklistService iEventTasklistService;

    @Override
    public Page<Map<String, Object>> get_seckill_skus(Page<?> page, QueryWrapper<?> qw) {
        var p = (Page<Map<String, Object>>) page;
        return p.setRecords(baseMapper.get_seckill_skus(page, qw));
    }

    @Override
    public List<Map<String, Object>> get_seckill_skus(QueryWrapper<?> qw) {
        return baseMapper.get_seckill_skus(null, qw);
    }

    //TODO 如果在redis中存储状态,应该是:0没有当前数据,1当前正在写入数据,2数据写入成功,3正在写入错误,4错误写入成功
    @Override
    @Transactional
    public void sync_send_msg() {
        log.info("开始锁一行");
        var v = Long.toString(Thread.currentThread().threadId()) + "_" + Long.toString(UUID.randomUUID().getLeastSignificantBits());
        if (!lock.opsForValue().setIfAbsent(RedisC.Lock.msg_sender, v, 30L, TimeUnit.SECONDS)) {
            log.info("获取锁 {} 失败,任务退出", RedisC.Lock.msg_sender);
            return;
        }
        var work_id = lock.opsForValue().increment(RedisC.Lock.work_id_gen);
        shared.put(work_id, new HashSet<>());
        Long cur_index = 0L;
        redisTemplate.opsForValue().setIfAbsent(RedisC.Shangjia.msg_sender_status, cur_index);
        cur_index = redisTemplate.opsForValue().get(RedisC.Shangjia.msg_sender_status);
        assert (Objects.nonNull(cur_index));
//        if(cur_index==null){
//            log.error("不存在key:{}",RedisC.Shangjia.msg_sender_status);
//            return;
//        }

        var psize = RedisC.Shangjia.msg_sender_batch_size;
        var p = 1L;
        var offset = 0;
        var base = cur_index / psize;
        p = base;
        if (base * psize == cur_index) {
            p++;
        } else {
            assert (cur_index > base * psize);
            offset = (int) (cur_index - base * psize);//安全的转换
        }

        var mapPage = new Page<Map<String, Object>>(p, psize);

        var q = new QueryWrapper<SeckillSession>();
        q.select("t1.id,t2.id");
        q.eq("t2.status", 1);
        q.orderByAsc("t1.id");

        var dd = get_seckill_skus(mapPage, q);
        while (!dd.getRecords().isEmpty()) {
            dd.getRecords().subList(offset, dd.getRecords().size()).forEach(i -> {
                rabbitTemplate.convertAndSend(AmqpC.Shangjia.msg_sender_exchange, AmqpC.Shangjia.msg_sender_key, i);
            });
            p++;
            offset = 0;
            dd = get_seckill_skus(mapPage, q);
        }

        unlock(v);
    }

    //TODO 分发任务的协议.这里简单的把每个上架后需要修改的记录的id传过去
    @Override
//    @Transactional
    public void sync_send_msg2() {
        log.info("尝试获取redisson锁");
        var lock = redissonClient.getLock(RedisC.Lock.Redisson.msg_sender);
        lock.lock(60 * 60, TimeUnit.SECONDS);
        log.info("取得锁 {}", RedisC.Lock.Redisson.msg_sender);

        var event_time = LocalDateTime.of(LocalDate.now(), LocalTime.of(0, 0, 0));
        var res = eventMapper.selectList(new QueryWrapper<Event>()
                .eq("event_type", 1)
                .eq("event_time", event_time)
        );
        if (res.isEmpty()) {
            Event entity = new Event();
            entity.setEvent_type(1);
            entity.setEvent_time(event_time);
            entity.setCreate_time(LocalDateTime.now());
            entity.setTitle("秒杀上架task发送");
            entity.setStatus(1);
            entity.setDescription("每天24:00:00开始上架当前设置好的秒杀活动的商品");
            eventMapper.insert(entity);
            res = eventMapper.selectList(new QueryWrapper<Event>().eq("event_type", 1).eq("event_time", event_time));
            assert (!res.isEmpty());
        }
        assert (res.size() == 1);
        var event = res.get(0);

        boolean can_run = false;
        if (event.getStatus() == 3) {
            log.info("任务已完成,退出");
        } else if (event.getStatus() == 2) {
            log.warn("当前任务被中断,无法进行!");
        } else {
            can_run = true;
        }

        if (can_run) {
            int p = 1, psize = 3;

            //处理未确认的发送
            var unacked_page = new Page<EventTasklist>(p, psize);

            var eq = new QueryWrapper<EventTasklist>()
                    .eq("e_id", event.getId())
                    .eq("acked", 0);
            unacked_page = iEventTasklistService.page(unacked_page, eq
            );
            try {
                while (!unacked_page.getRecords().isEmpty()) {
                    unacked_page.getRecords().forEach(i -> {
                        var m = new HashMap<>();
                        m.put("range_a", i.getRange_a());
                        m.put("range_b", i.getRange_b());
                        rabbitTemplate.convertAndSend(AmqpC.Shangjia.msg_sender_exchange, AmqpC.Shangjia.msg_sender_key, m);
                    });
                    iEventTasklistService.update(new UpdateWrapper<EventTasklist>()
                            .setSql("acked=1")
                            .in("id", unacked_page.getRecords().stream().map(EventTasklist::getId).toList())
                    );
                    iEventTasklistService.page(unacked_page, eq);
                }
            } catch (AmqpException e) {
                log.error("未确认的EventTasklist处理出错:mq发送失败==>{}", e);
                throw e;
            } catch (Exception e) {
                log.error("未确认的EventTasklist处理出错:{}", e);
                throw e;
            }

            log.info("未确认的EventTasklist处理完毕");

            var eventTasklists = iEventTasklistService.list(new QueryWrapper<EventTasklist>()
                    .eq("e_id", event.getId())
                    .apply("""
                            exists( 
                            select max(t2.range_b) as bb 
                                from sms_event_tasklist t2 
                            where 
                                t2.acked=1 and t2.e_id=sms_event_tasklist.e_id 
                            HAVING 
                                range_b  =bb 
                            )
                            """)
            );
            long next_range_a = 0;
            //首次开始派发活动
            if (eventTasklists.isEmpty()) {
                //获取初始的左边界
                var list = get_seckill_skus(new QueryWrapper<>()
                        .select("t2.id as id")
                        .ge("t1.start_time", event_time)
                        .le("t1.start_time", event_time.plusDays(3))
                        .eq("t1.status", 1)
                        .orderByAsc("t1.id", "t2.id")
                        .last("limit 1"));
                if (list.isEmpty()) {
                    //没有活动需要派发,所以下面直接找不到事件进行发送
                    //TODO 结束逻辑
//                    event.setFinshed_time(LocalDateTime.now());
//                    event.setStatus(3);
//                    eventMapper.updateById(event);
//                    assert (false);
                    next_range_a = 0;
                } else {
                    assert (list.size() == 1);
                    next_range_a = Long.parseLong(list.get(0).get("id").toString());
                }
            } else {//有人派发了,直接接着派发
                next_range_a = eventTasklists.get(0).getRange_b();
            }

            var old_range_a = next_range_a;

            p = 1;
            var q = new QueryWrapper<>();
            q.select("t2.id as id")
                    .ge("t1.start_time", event_time)
                    .le("t1.start_time", event_time.plusDays(3))
                    .eq("t1.status", 1)
                    .ge("t2.id", next_range_a)
                    .orderByAsc("t1.id", "t2.id");
            var page = new Page<Map<String, Object>>(p, psize);
            get_seckill_skus(page, q);

            try {
                while (!page.getRecords().isEmpty()) {

                    next_range_a = Long.parseLong(page.getRecords().get(page.getRecords().size() - 1).get("id").toString());

                    EventTasklist tasklist = new EventTasklist();
                    tasklist.setAcked(0);
                    tasklist.setRange_a(old_range_a);
                    tasklist.setRange_b(next_range_a);
                    tasklist.setE_id(event.getId());
                    tasklist.setCreate_date(LocalDateTime.now());
                    iEventTasklistService.save(tasklist);
                    var m = new HashMap<>();
                    m.put("range_a", old_range_a);
                    m.put("range_b", next_range_a);
                    var id=redisTemplate.opsForValue().increment(RedisC.Shangjia.correlation_id);
                    var corrd=new CorrelationData(id.toString());
                    rabbitTemplate.convertAndSend(AmqpC.Shangjia.msg_sender_exchange, AmqpC.Shangjia.msg_sender_key, m,corrd);
                    CorrelationData.Confirm confirm = corrd.getFuture().get();
                    if(confirm.isAck()) {
                        log.info("发送一条事件:[{}-{})", old_range_a, next_range_a);
                    }else{
                        throw new RuntimeException("消息发送失败,caused by ,消息未被确认");
                    }

                    //TODO 应该在callbackconfirm里设置的,这里先这样
                    EventTasklist tasklist1 = new EventTasklist();
                    tasklist1.setId(tasklist.getId());
                    tasklist1.setAcked(1);
                    iEventTasklistService.updateById(tasklist1);

                    if (old_range_a == next_range_a) {
                        log.info("事件发送完毕!");
                        break;
                    }

                    old_range_a = next_range_a;
                    //FIXME 清理原来的page或者new一个新的page
                    get_seckill_skus(page, new QueryWrapper<>()
                            .select("t2.id as id")
                            .ge("t1.start_time", event_time)
                            .le("t1.start_time", event_time.plusDays(3))
                            .eq("t1.status", 1)
                            .ge("t2.id", next_range_a)
                            .orderByAsc("t1.id", "t2.id"));

                }
            } catch (AmqpException e) {
                log.error("消息发送失败: {}", e);
                throw e;
            }/*finally {
            try {
                lock.unlock();
            }catch (IllegalMonitorStateException e){
                log.error("释放锁失败: {} {}",RedisC.Lock.Redisson.msg_sender,e);
                throw e;
                //数据库回滚操作
            }
        }*/ catch (ExecutionException e) {
                throw new RuntimeException(e);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            event.setFinshed_time(LocalDateTime.now());
            event.setStatus(3);
            eventMapper.updateById(event);
        }

        //TODO 不放到finally里是为了分辨出两个不同的异常
        try {
            lock.unlock();
        } catch (IllegalMonitorStateException e) {
            log.error("释放锁失败: {} {}", RedisC.Lock.Redisson.msg_sender, e);
            throw e;
        }
    }

    @Deprecated
    public void unlock(String v) {
        lock.execute(new DefaultRedisScript<>("""
                if redis.call("get",KEYS[1]) == ARGV[1]
                then
                    return redis.call("del",KEYS[1])
                else
                    return 0
                end
                """), List.of(RedisC.Lock.msg_sender), v);
        log.info("释放一行锁");
    }

    @Override
    public void test_redis_atomic() {
        var left_num = redisTemplate.opsForValue().decrement("num");
        if (left_num >= 0) {
            redisTemplate.opsForValue().increment("buy");
        }
    }

    private static int get_shangjia_lock(LockMapper lockMapper) {
        var oldv = lockMapper.get_lock_version();
        var res = lockMapper.update_lock_version(oldv);
        return res - 1;
    }
}
